package com.qs.rocketmq.bt.mq.batch;

import com.qs.rocketmq.bt.mq.MqConstant;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.ArrayList;
import java.util.List;

public class BatchMsgProducer {

    private static final DefaultMQProducer producer;

    static {
        // 实例化消息生产者Producer
        producer = new DefaultMQProducer(MqConstant.PRODUCER_GROUP_NAME);
        // 设置NameServer的地址
        producer.setNamesrvAddr(MqConstant.MQ_NAME_SRV_ADDR);
    }

    public static void main(String[] args) throws Exception {
        // batchSend();
        batchSendForLargeMsg();
    }

    /**
     * 批量消息一个批次不能超过4MB，如果确定发的消息内容达不到4MB，则直接发就好了
     */
    public static void batchSend() throws Exception {
        // 启动Producer实例
        producer.start();

        String topic = MqConstant.BATCH_MSG_TOPIC;
        List<Message> messages = new ArrayList<>();
        messages.add(new Message(topic, "TagA", "OrderID001", "Hello world 0".getBytes()));
        messages.add(new Message(topic, "TagA", "OrderID002", "Hello world 1".getBytes()));
        messages.add(new Message(topic, "TagA", "OrderID003", "Hello world 2".getBytes()));
        try {
            producer.send(messages);
        } catch (Exception e) {
            e.printStackTrace();
        }

        // 如果不再发送消息，关闭Producer实例。
        producer.shutdown();
    }

    /**
     * 消息内容比较大的时候，那么需要将消息进行拆分，分批次发送，可以使用 ListSplitter 类将消息分割后发送
     */
    public static void batchSendForLargeMsg() throws MQClientException {
        producer.start();

        // 组装消息
        String topic = MqConstant.BATCH_MSG_TOPIC;
        List<Message> messages = new ArrayList<>();
        messages.add(new Message(topic, "TagA", "OrderID001", "Hello world 0".getBytes()));
        messages.add(new Message(topic, "TagA", "OrderID002", "Hello world 1".getBytes()));
        messages.add(new Message(topic, "TagA", "OrderID003", "Hello world 2".getBytes()));
        messages.add(new Message(topic, "TagA", "OrderID004", "Hello world 3".getBytes()));
        messages.add(new Message(topic, "TagA", "OrderID005", "Hello world 4".getBytes()));
        messages.add(new Message(topic, "TagA", "OrderID006", "Hello world 5".getBytes()));
        messages.add(new Message(topic, "TagA", "OrderID007", "Hello world 6".getBytes()));
        messages.add(new Message(topic, "TagA", "OrderID008", "Hello world 7".getBytes()));
        messages.add(new Message(topic, "TagA", "OrderID009", "Hello world 8".getBytes()));

        //把大的消息分裂成若干个小的消息
        ListSplitter splitter = new ListSplitter(messages);
        while (splitter.hasNext()) {
            try {
                List<Message>  listItem = splitter.next();
                producer.send(listItem);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        producer.shutdown();
    }

}
